package com.dis.fan;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink CDC 2.2.1, 捕获Oracle数据表变化，并写入MySQL
 *
 * @author 闻武
 * @since 2020-05-31
 */
//todo: 目前有问题 20220601
public class CdcOracle2Oracle {
    public static void main(String[] args) throws Exception {
        /**
         * 1 > 程序入口
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.disableOperatorChaining();

        /**
         * 2> 输入参数，可参考： -hostname 192.168.1.197 -port 1521 -username flinkcdc -password flinkcdc -databaseName orcl -schemaName flinkcdc
         * （Idea环境，Run ---> Edit Configurations，打开Run/Debug Configurations，在相关工程的Program arguments中输入）
         */
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostname = parameterTool.get("hostname");
        int port = parameterTool.getInt("port");
        String username = parameterTool.get("username");
        String password = parameterTool.get("password");
        String databaseName = parameterTool.get("databaseName");
        String schemaName = parameterTool.get("schemaName");

        /**
         * 3> 流表
         */
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String strSourceSql = "CREATE TABLE IF NOT EXISTS products_ora_cdc (\n" +
                "    ID INT,\n" +
                "    NAME STRING,\n" +
                "    DESCRIPTION STRING,\n" +
                "    PRIMARY KEY (ID) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'oracle-cdc',\n" +
                "    'hostname' = '" + hostname + "',\n" +
                "    'port' = '" + port + "',\n" +
                "    'username' = '" + username + "',\n" +
                "    'password' = '" + password + "',\n" +
                "    'database-name' = '" + databaseName + "',\n" +
                "    'schema-name' = '" + schemaName + "',  \n" +
                "    'table-name' = 'products',\n" +
                "    'debezium.log.mining.continuous.mine'='true',\n" +
                "    'debezium.log.mining.strategy'='online_catalog',\n" +
                "    'debezium.database.tablename.case.insensitive'='false',\n" +
                "    'scan.startup.mode' = 'initial')";

        String strSinkSql = " CREATE TABLE IF NOT EXISTS products_ora_sink (" +
                "    ID INT,\n" +
                "    NAME STRING,\n" +
                "    DESCRIPTION STRING,\n" +
                "    PRIMARY KEY (ID) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'jdbc'," +
                "    'driver' = 'oracle.jdbc.OracleDriver'," +
                "    'url' = 'jdbc:oracle:thin:@192.168.1.212:1521:orcl'," +
                "    'table-name' = 'products'," +
                "    'username' = 'cdctest'," +
                "    'password' = 'cdctest' " +
//                "    'url' = 'jdbc:oracle:thin:@" + hostname + ":" + port + ":" + databaseName + "'," +
//                "    'table-name' = 'products_sink'," +
//                "    'username' = '" + username + "'," +
//                "    'password' = '" + password + "' " +
                " )";

        tableEnv.executeSql(strSourceSql);
        tableEnv.executeSql(strSinkSql);
        tableEnv.executeSql("insert into products_ora_sink(ID, NAME, DESCRIPTION) select ID, NAME, DESCRIPTION from products_ora_cdc ");
    }
}